package com.baishancloud.log.live.sanma

import com.baishancloud.log.common.util.{StringUtil, TimeUtil}
import com.baishancloud.log.format.java.util.IPLocator
import com.baishancloud.log.format.scala.SanmaOrigin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.util
import java.util.Objects


class SanmaLogRecordParse(parameterTool: ParameterTool) extends ProcessFunction[SanmaOrigin, SanmaLog] {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[SanmaLogRecordParse])

  override def processElement(value: SanmaOrigin, ctx: ProcessFunction[SanmaOrigin, SanmaLog]#Context, out: Collector[SanmaLog]): Unit = {
    val ipInfo = IPLocator.lookup(value.cIp())

    var isp = ipInfo.getIsp.split("/")(0)
    if (isp.equals("other")) isp = "其他"
    if (isp.contains("广电")) {
      isp = "广电"
    } else if (isp.contains("华数")) {
      isp = "华数"
    }
    val country = if (ipInfo.getCountry.equals("other")) "其他" else ipInfo.getCountry
    var province = if (ipInfo.getProvince.equals("other")) "其他" else ipInfo.getProvince
    if ((country.equals("中国") || country.startsWith("CHINA")) && province.equals("其他")) province = "中国其他地区"

    val complexFieldList: util.List[String] = StringUtil.split(value.complexField1(), '@')
    val uv: Int = if (complexFieldList.size() >= 5) {
      if (Objects.equals(complexFieldList.get(4), "1")) {
        1
      } else {
        0
      }
    } else {
      0
    }

    val logRecord: SanmaLog = SanmaLog(
      TimeUtil.timeAtString(value.datetimeOfUTC().toEpochSecond, parameterTool.getLong(timeAtSize, 1) * 60),
      StringUtil.subString(value.xvHost(), 32),
      StringUtil.subString(country, 32),
      StringUtil.subString(province, 32),
      StringUtil.subString(isp, 32),
      0,
      uv
    )
    out.collect(logRecord)
  }
}
